package krasa;

import java.util.concurrent.DelayQueue;
import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;

/**
 * @author Vojtech Krasa
 */
public class DelayQueueExecutionRateLimiter extends ExecutionRateLimiter {
	protected final Thread thread;
	DelayQueue<ThreadWakeUp> delayQueue = new DelayQueue<ThreadWakeUp>();

	public DelayQueueExecutionRateLimiter(ExecutionRateLimiterConfiguration configuration) {
		super(configuration);
		thread = new Thread(new Runnable() {
			@Override
			public void run() {
				while (true) {
					Delayed poll = null;
					try {
						if (delayQueue.isEmpty()) {
							continue;
						}
						poll = delayQueue.take();
						synchronized (poll) {
							poll.notify();
						}
					} catch (Exception e) {
						e.printStackTrace();
					}
				}
			}
		}, "DelayQueuePoller");
		thread.start();

	}

	@Override
	public void run() {
		ThreadWakeUp e = new ThreadWakeUp(getWakeUpTime());
		delayQueue.offer(e);
		synchronized (e) {
			try {
				e.wait();
			} catch (InterruptedException e1) {
				System.err.println("InterruptedException " + Thread.currentThread().getName() + " " + System.currentTimeMillis());
			}
		}

	}


	private synchronized long getWakeUpTime() {
		long deltaInMillis = configuration.getDeltaInMillis();
		long next = state.getLastExecutionTimeInMillis() + deltaInMillis;
		long current = System.currentTimeMillis();
		if (next - current > 0) {
			state.setLastExecutionTimeInMillis(next);
		} else {
			state.setLastExecutionTimeInMillis(current);
		}
		return next;
	}


	class ThreadWakeUp implements Delayed {
		private Long wakeUpTime;

		public ThreadWakeUp(long wakeUpTime) {
			this.wakeUpTime = wakeUpTime;
		}

		@Override
		public long getDelay(TimeUnit unit) {
			return unit.convert(wakeUpTime - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
		}

		@Override
		public int compareTo(Delayed o) {
			Long aLong = getDelay(TimeUnit.MILLISECONDS);
			return wakeUpTime.compareTo(aLong);
		}

	}

}
